package org.shj.spark.dataframe;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

import scala.Tuple2;

/**
 * ./bin/spark-submit --master spark://192.168.1.23:7070 --class org.shj.spark.dataframe.JDBCDataset
 * --driver-class-path ./lib/mysql-connector-java-5.1.32.jar
 * --jars ./lib/mysql-connector-java-5.1.32.jar
 * test.jar
 * 
 * 如果使用 standalone cluster 模式来运行，配置 spark-env.sh ：
 * export SPARK_CLASSPATH=/opt/programs/spark-2.0-bin-hadoop-2.7/lib/mysql-connector-java-5.1.8-bin.jar
 * ./bin/spark-submit --master spark://192.168.1.23:7070 --deploy-mode-cluster --class org.shj.spark.dataframe.JDBCDataset 
 * hdfs://node22.8020/test.jar
 * 
 * ./bin/spark-submit --master yarn-client --class org.shj.spark.dataframe.JDBCDataset ./test.jar
 * 
 * 在conf/spark-default.conf 里面配置：
 * spark.driver.extraClassPath=/opt/programs/spark-2.0-bin-hadoop-2.7/lib/mysql-connector-java-5.1.8-bin.jar
 * spark.executor.extraClassPath=/opt/programs/spark-2.0-bin-hadoop-2.7/lib/mysql-connector-java-5.1.8-bin.jar
 * 
 * 
 * @author Administrator
 *
 */
public class JDBCDataset {

	public static void main(String[] args) {
		SparkSession ss = SparkSession.builder().master("local").
				appName("JDBCDataset").getOrCreate();

		Map<String, String> options = new HashMap<String,String>();
		options.put("url", "jdbc:mysql://192.168.31.227:3306/test");
		options.put("dbtable", "student");
		options.put("user", "root");
		options.put("password", "123456");
		
		Dataset<Row> students = ss.read().format("jdbc").options(options).load();
		
		options.put("dbtable", "student_score");
		Dataset<Row> scores = ss.read().format("jdbc").options(options).load();
		
		JavaPairRDD<String, Tuple2<Integer, Integer>> studentInfos = students.javaRDD().mapToPair(new PairFunction<Row, String, Integer>() {
			private static final long serialVersionUID = 1L;

			@Override
			public Tuple2<String, Integer> call(Row row) throws Exception {
				return new Tuple2<String,Integer>(row.getAs("name"), Integer.parseInt(row.getAs("age")));
			}
		}).join(scores.javaRDD().mapToPair(new PairFunction<Row, String, Integer>() {

			private static final long serialVersionUID = -608303144936137781L;

			@Override
			public Tuple2<String, Integer> call(Row row) throws Exception {
				return new Tuple2<String,Integer>(row.getAs("name"), Integer.parseInt(row.getAs("score")));
			}
		}));
		
		JavaRDD<Row> filter = studentInfos.map(new Function<Tuple2<String, Tuple2<Integer, Integer>>, Row>() {
			private static final long serialVersionUID = 1L;

			public Row call(Tuple2<String, Tuple2<Integer, Integer>> tuple) throws Exception {
				return RowFactory.create(tuple._1, tuple._2._1, tuple._2._2);
			}
		}).filter(new Function<Row, Boolean>() {
			private static final long serialVersionUID = 1L;

			public Boolean call(Row row) throws Exception {
				return row.getInt(2) > 80;
			}
		});
		
		//RDD 转成 Dataset 
		List<StructField> list = new ArrayList<StructField>();
		list.add(DataTypes.createStructField("name", DataTypes.StringType, true));
		list.add(DataTypes.createStructField("age", DataTypes.IntegerType, true));
		list.add(DataTypes.createStructField("score", DataTypes.IntegerType, true));
		
		StructType schema = DataTypes.createStructType(list);
		Dataset<Row> stuDF = ss.sqlContext().createDataFrame(filter, schema);
		for(Row row : stuDF.collectAsList()){
			System.out.println(row);
		}
		
		stuDF.javaRDD().foreach(new VoidFunction<Row>() {
			private static final long serialVersionUID = -4460111391424242549L;

			public void call(Row t) throws Exception {
				System.out.println(t);
			}
		});
		
		ss.close();

	}

}
